package com.wjk.kylin.mall.product.kafka;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.wjk.kylin.mall.common.core.constant.Constants;
import com.wjk.kylin.mall.product.dto.CanalKafkaMessageDTO;
import com.wjk.kylin.mall.product.entity.es.ProductESBean;
import com.wjk.kylin.mall.product.enums.ProductDataSyncEsTableEnum;
import com.wjk.kylin.mall.product.es.ProductEsServiceImpl;
import com.wjk.kylin.mall.product.kafka.datasync.ProductDataSyncExecutor;
import com.wjk.kylin.mall.product.kafka.datasync.ProductDataSyncExecutorFactory;
import com.wjk.kylin.mall.product.service.IProductService;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.*;

/**
 * binlog -> canal -> kafka -> 消费者 -> es
 * 商品同步es
 * @author wjk
 */
@Slf4j
@Component
public class ProductConsumer {

    @Autowired
    private IProductService productService;
    @Autowired
    private ProductEsServiceImpl productEsService;

    @Autowired
    private MeterRegistry registry;
    private Counter productToEsErrorCounter;

    @PostConstruct
    private void init() {
        productToEsErrorCounter = registry.counter("mall_product_data_sync_es_exception_total");
    }

    @KafkaListener(topics = {"bd_canal_mall_product_es"})
    public void productDataSyncEsListener(List<ConsumerRecord<String, String>> list, Acknowledgment ack) {
        long start = System.currentTimeMillis();
        MDC.put(Constants.TRACE_ID_KEY, IdUtil.simpleUUID());
        try {
            if (CollectionUtil.isNotEmpty(list)) {
                List<CanalKafkaMessageDTO> dtoList = new ArrayList<>();
                List<Long> productIdList = new ArrayList<>();
                //1.解析消息
                for (ConsumerRecord<String, String> record : list) {
                    log.info("ProductConsumer-goodsDataSyncListener,topic:{},partition:{},offset:{},key:{},value:{}",
                            record.topic(), record.partition(), record.offset(), record.key(), record.value());

                    CanalKafkaMessageDTO messageDTO = handleKafkaLogsToMessage(record.value(), productIdList);
                    if (messageDTO != null) {
                        dtoList.add(messageDTO);
                    }
                }
                log.info("ProductConsumer-productDataSyncEsListener size:{},解析数据耗时:{}ms", list.size(), (System.currentTimeMillis() - start));
                start = System.currentTimeMillis();

                //1.查询mysql
                List<Long> existProductIdList = new ArrayList<>();
                if (CollectionUtil.isNotEmpty(productIdList)) {
                    existProductIdList = productService.getProductIdListByIDList(productIdList);
                }
                //需要删除es的商品id
                List<Long> deleteIdList = new ArrayList<>();
//                for (Long productId : productIdList) {
//                    if (!existProductIdList.contains(productId)) {
//                        deleteIdList.add(productId);
//                    }
//                }
                //2.查询es
                Map<Long, ProductESBean> isExistEsMap = new HashMap<>();//key:id,value:esBeanList
                if (CollectionUtil.isNotEmpty(existProductIdList)) {
                    //不存在于db的商品id
                    for (Long productId : productIdList) {
                        if (!existProductIdList.contains(productId)) {
                            deleteIdList.add(productId);
                        }
                    }
                    isExistEsMap = productEsService.getProductEsMap(existProductIdList);
                } else {
                    //都不存在于db，就是全删除
                    deleteIdList = productIdList;
                }

                log.info("ProductConsumer-productDataSyncEsListener size:{},查询mysql和es耗时:{}ms", list.size(), (System.currentTimeMillis() - start));
                start = System.currentTimeMillis();

                //
                if (CollectionUtil.isNotEmpty(dtoList)) {
                    for (CanalKafkaMessageDTO dto : dtoList) {
                        handleKafkaLogs(dto, isExistEsMap, deleteIdList);
                    }
                }
                log.info("ProductConsumer-productDataSyncEsListener size:{},组装es耗时:{}ms", list.size(), (System.currentTimeMillis() - start));
                start = System.currentTimeMillis();

                //新增、更新、删除es操作
                List<ProductESBean> insertOrUpdateEsList = null;
                if (CollectionUtil.isNotEmpty(isExistEsMap)) {
                    insertOrUpdateEsList = Lists.newArrayList(isExistEsMap.values());
                }

                productEsService.bulkProductEsList(insertOrUpdateEsList, deleteIdList);

                log.info("ProductConsumer-productDataSyncEsListener size:{},保存es耗时:{}ms", list.size(), (System.currentTimeMillis() - start));
            }
            ack.acknowledge();
        } catch (Exception e) {
            productToEsErrorCounter.increment();
            if (CollectionUtil.isNotEmpty(list)) {
                for (ConsumerRecord<String, String> record : list) {
                    log.error("ProductConsumer-goodsDataSyncListener,topic:{},partition:{},offset:{},key:{},value:{}",
                            record.topic(), record.partition(), record.offset(), record.key(), record.value());
                }
            }
            log.error("ProductConsumer-goodsDataSyncListener,e:", e);
        } finally {
            MDC.remove(Constants.TRACE_ID_KEY);
        }
    }

    private void handleKafkaLogs(CanalKafkaMessageDTO dto, Map<Long, ProductESBean> isExistEsMap, List<Long> deleteIdList) {
        String tableName = dto.getTable().toLowerCase();
        String actionType = dto.getType();
        ProductDataSyncExecutor executor = ProductDataSyncExecutorFactory.getExecutor(tableName);
        if (executor == null) {
            log.error("DataSyncGoodsConsumer.handleKafkaLogs-查询执行器为空,tableName:{}", tableName);
            return;
        }
        switch (actionType) {
            case "INSERT":
                executor.insert(dto, isExistEsMap, deleteIdList);
                break;
            case "UPDATE":
                executor.update(dto, isExistEsMap, deleteIdList);
                break;
            case "DELETE":
                executor.delete(dto, isExistEsMap, deleteIdList);
                break;
            default:
                log.info("DataSyncGoodsConsumer.handleKafkaLogs,tableName:{},actionType:{} do nothing...", tableName, actionType);
                break;
        }
    }

    /**
     * @param json
     * @param productIdList
     * @return
     */
    private CanalKafkaMessageDTO handleKafkaLogsToMessage(String json, List<Long> productIdList) {
        if (StringUtils.isEmpty(json)) {
            return null;
        }
        CanalKafkaMessageDTO dto = JSONObject.parseObject(json, CanalKafkaMessageDTO.class);
        if (dto == null) {
            return null;
        }
        String tableName = dto.getTable();
        String actionType = dto.getType();
        List<JSONObject> dataList = dto.getData();
        List<String> pkNames = dto.getPkNames();
        if (StringUtils.isBlank(tableName) || StringUtils.isBlank(actionType) || CollectionUtil.isEmpty(dataList) || CollectionUtil.isEmpty(pkNames)) {
            log.error("ProductConsumer-handleKafkaLogs table or type or data or pkNames is blank,json:{}", json);
            return null;
        }
        tableName = tableName.toLowerCase();
        ProductDataSyncEsTableEnum tableEnum = ProductDataSyncEsTableEnum.getByValue(tableName);
        if (tableEnum == null) {
            log.error("ProductConsumer-handleKafkaLogs ProductDataSyncEsTableEnum is blank,tableName:{}", tableName);
            return null;
        }
        for (JSONObject data : dataList) {
            Object productIdObj = data.get(tableEnum.getProductIdField());
            if (productIdObj != null) {
                productIdList.add(Long.parseLong(productIdObj.toString()));
            } else {
                log.error("ProductConsumer-handleKafkaLogs productIdObj is blank,tableName:{},goodsIdField():{}", tableName, tableEnum.getProductIdField());
            }
        }
        return dto;
    }
}
